from typing import Iterable

import pandas as pd
from pyspark import Row
from pyspark.sql import SparkSession
from spark_utils import spark_rows_to_df
from pyspark.sql.functions import *
from pyspark.sql.types import *

# windows下需要此代码
import findspark
findspark.init()

# 创建spark环境:sparkSession
spark = SparkSession.builder.appName("test").master("local[1]").getOrCreate()

# 设置日志级别:ALL, DEBUG, ERROR, FATAL, INFO, OFF, TRACE, WARN
spark.sparkContext.setLogLevel("INFO")

# 使用spark读取数据源示例(分布式读取，一个文件(文件块)会生成一个task )
spark_df = (spark.read.option("header","true"). # header，代表将csv的表头作为数据表的列名
            csv("C:\\Users\\ljc\\Downloads\\6cff080c-945e-4657-8055-f7bf00f23d8a.csv"))


# 模拟partition内部转换操作，返回二维列表
def my_map(rows:Iterable[Row]):
    pd_df = spark_rows_to_df(rows)
    error_file = []
    error_file.append([1001,'sn00001','starttime1','endtime1','2023-01-01'])
    error_file.append([1002,'sn00002', 'starttime2','endtime2', '2023-02-01'])
    return error_file

# 写出时的表结构
schema = StructType([
    StructField('id',IntegerType()),
    StructField('sn',StringType()),
    StructField('start_time',StringType()),
    StructField('end_time',StringType()),
    StructField('p_date',StringType())
])

# spark流程
spark_df.rdd.mapPartitions(my_map).toDF(schema).repartition(1).write.partitionBy("p_date").option("header","true").mode("Append").csv("s3://地址")


# spark_df.show()
# 转换为pandas on spark dataframe
# spark_df.to_pandas_on_spark()
#
# # 先读取为pandas dataframe，再转换成spark dataframe示例
# p = pd.read_csv('C:\\Users\\ljc\\Downloads\\6cff080c-945e-4657-8055-f7bf00f23d8a.csv')
# spark.createDataFrame(p).show()
#
# # spark读取数据，使用DataFrame API示例
# ((spark.read.option("header","true"). # header，代表将csv的表头作为数据表的列名
#             csv("C:\\Users\\ljc\\Downloads\\6cff080c-945e-4657-8055-f7bf00f23d8a.csv")).
#             createOrReplaceTempView("mytable"))
#
# # SQL示例，show()打印结果至控制台,createOrReplaceTempView将数据创建为表
# (spark.sql(" select * from mytable where sn = '1111111' ").
#  createOrReplaceTempView("writetable")) #createOrReplaceTempView代表将数据创建为表，参数为表名，在之后可以使用sql读取此表
#
# # SQL示例，将数据按照DataFrame格式返回
# spark_df2 = spark.sql(" select * from mytable where sn = '1111111' ")
# spark_df2.show()
#
# # spark DataFrame计算示例
# count_df = spark_df2.groupBy("sn").count
# count_df.show()
#
# # spark DataFrame写出示例
# spark_df.write.csv("s3://my_bucket/my_csv")
# spark_df.write.text("s3://my_bucket/my_csv")
# # 写出时通过p_date分区，每个值会存在不同的p_date文件夹下
# spark_df.write.partitionBy("p_date").parquet("s3://my_bucket/my_csv")
#
# # spark1版本 RDD示例
# # map操作：传入每一条数据(由于rdd是DataFrame转化而来，所以是row格式)，返回一条数据
# def my_map(row:Row):
#     v1 = row[1]
#     v2 = row[3]
#     return v1 + v2
# spark_df.rdd.map(my_map).filter(lambda x: x>1).foreach(print)
#




# 停止spark环境
spark.stop()
